package com.zsp.coupon.service.imp;

import com.alibaba.fastjson.JSON;
import com.zsp.coupon.constant.Constant;
import com.zsp.coupon.constant.CouponStatus;
import com.zsp.coupon.dao.CouponDao;
import com.zsp.coupon.entity.Coupon;
import com.zsp.coupon.service.IKafkaService;
import com.zsp.coupon.vo.CouponKafakMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * kafak相关的服务接口
 * 核心思想：将cache中的Coupon状态同步到数据库当中
 */

@Component
@Slf4j
public class KafkaServiceImpl implements IKafkaService {

/** coupondao接口*/

    private CouponDao couponDao;
    /**
     * 消费优惠卷kAFAK消息
     * @param record {@link ConsumerRecord}
     */
    @Override
    @KafkaListener(topics = {Constant.TOPIC},groupId = "imooc-coupon-1")
    public void consumeCouponKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafaMessage = Optional.ofNullable(record.value());
        if (kafaMessage.isPresent()) {
            Object message = kafaMessage.get();
            CouponKafakMessage couponInfo = JSON.parseObject(
                    message.toString(),
                   CouponKafakMessage.class
            );
            log.info("Revice Couponkafak:{}",message.toString());

            CouponStatus status = CouponStatus.of(couponInfo.getStataus());

            switch(status){
                case USEABLE:
                    break;
                case USED:
                    processUsedCoupons(couponInfo,status);
                    break;
                case EXPIRED:
                    processExpiredCoupons(couponInfo,status);
                    break;
            }
        }
    }

    /**
     * 处理已经使用的优惠卷
     * @param kafakMessage
     * @param status
     */
    private void  processUsedCoupons(CouponKafakMessage kafakMessage,
                                     CouponStatus status  ) {
        // TODD 给用户发送短信
                    processCouponByStatus(kafakMessage,status);
    }

    /**
     * 处理过期的优惠卷
     * @param kafakMessage
     * @param status
     */
    private void processExpiredCoupons(CouponKafakMessage kafakMessage,
                                       CouponStatus status ) {
        processCouponByStatus(kafakMessage,status);

    }

    /**
     * 根据状态处理优惠卷信息
     * @param kafakMessage
     * @param status
     */

    private void  processCouponByStatus(CouponKafakMessage kafakMessage,
                                        CouponStatus status) {
        List<Coupon> coupons = couponDao.findAllById(kafakMessage.getIds());
        if (CollectionUtils.isEmpty(coupons)
            || coupons.size() != kafakMessage.getIds().size()) {
            log.error("can not find right coupon info:{}",
                    JSON.toJSONString(kafakMessage));
            return;
        }
        coupons.forEach( c -> c.setStatus(status));
        log.info("CouponKafakMessage OP Coupon count:{}",
                couponDao.saveAll(coupons).size());
    }
}
